/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ocr.util;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.logging.Level;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class SenderDemo implements Runnable {

    static Logger logger = Logger.getLogger(SenderDemo.class.getName());
    public static String confPath;
    public RESendSession session;

    public SenderDemo(RESendSession session) {
        this.session = session;
    }

    public static void main(String[] args) throws RESessionException, REConnectionException {
        //读取配置文件，linux下运行使用
//        SenderDemo.confPath = args[0];
//        PropertyConfigurator.configure(SenderDemo.confPath + "/log4j.properties");
        //生成连接，参数为rocketmq的nameserver串。可以多个nameserver，分号隔开。由具体业务决定

        REConnection conn = new REConnection("10.136.64.28:9877");

        //生成builder，参数为topic，由具体业务决定
        RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder("jiyutest_mq_1");
        //设置数据结构，参数为每一列的列名、列类型、是否可以为空。注意： 如果设置为不能为空，那么如果添加数据时没有添加该列数据，会报错。
        builder.addColumn("uid", REFieldType.String, true);
        builder.addColumn("ugid", REFieldType.String, true);
        builder.addColumn("ifexists", REFieldType.Strings, true);
        builder.addColumn("wb_lpic_name", REFieldType.Strings, true);

        /**
         * 这里是设置struct或者structs的格式。 首先需要addColumn
         * 其次需要在对应的类型里面setDetail，参数是：列名，子列类型map
         */
        //生成session
        ExecutorService executor = Executors.newFixedThreadPool(100);
        for (int i = 0; i < 1; i++) {
            RESendSession session = (RESendSession) builder.build();
            session.start();
            SenderDemo demo = new SenderDemo(session);
            Thread t = new Thread(demo);
            executor.execute(t);
        }
    }

    @Override
    public void run() {
        try {
            //开始发送。
            //这里，DocSchemaName和UserDesc非必须，具体什么时候需要，由具体业务决定。
            //DocSchemaName设置一次后不会清空，每次flush都是同一个，如果需要改变，需要再次设置。
            session.setDocSchemaName("doc_schema_name_all");
            //UserDesc设置后，只会在一次flush起作用，下一批数据发送需要重新设置。
            HashMap<String, String> user_desc = new HashMap<>();
//            user_desc.put("user", "test");
//            user_desc.put("pwd", "pwd");
//            user_desc.put("round", "all");
            int num = 1;
            for (int r = 0; r < 10; ++r) {
                session.setUserDesc(user_desc);
                //添加数据。set完所有列后add即可添加一条数据。多条数据打一包发送。
                //各种set方法可以通过session.set补全来查看。set方法和REFieldType各类对应。
                //注意：尽量不要一条数据发送一次，这样会带来较大的额外开销
                //注意2：本地缓存数据量有大小限制，如果超过500M，会报错。
                //注意3：如果要用set（index，value）的方式设置数据，index从1开始。
                try {
                    Thread.sleep(50);
                } catch (InterruptedException ex) {
                }
                for (int i = 0; i < 1000; ++i) {
                    num++;
                    session.setString("uid", "uid" + i);
                    session.setString("ugid", "ugid" + i);
//                    String[] str = {"0|test/test" + num, "0|test/tt" + num};
                    String[] str = {"0|test/test" + num};
                    session.setStrings("ifexists", str);
//                    String[] str1 = {"a/test" + num, "b/tt" + num};
                    String[] str1 = {"c/tt" + num};
                    session.setStrings("wb_lpic_name", str1);
                    session.add();
                    if (i % 3000 == 0) {
                        //获得各种统计量
                        //当前缓存的数据总量
                        logger.info("current data size : " + session.getCurrentDataSize());
                        //当缓存数据较多时，数据会被分割为小包，查看当前缓存的数据包数
                        logger.info("current pac num : " + session.getCurrentPacNum());
                        //查看当前数据包中数据的条数
                        logger.info("record num int current pac : " + session.getRecordNumInCurrentPac());
                    }
                }
                logger.info("current data size : " + session.getCurrentDataSize());
                logger.info("current pac num : " + session.getCurrentPacNum());
                logger.info("record num int current pac : " + session.getRecordNumInCurrentPac());
                logger.info(new Date().toLocaleString());
                //添加完所有数据，flush即可发送。
                session.flush();
            }
        } catch (RESessionException ex) {
            java.util.logging.Logger.getLogger(SenderDemo.class.getName()).log(Level.SEVERE, null, ex);
        } catch (REConnectionException ex) {
            java.util.logging.Logger.getLogger(SenderDemo.class.getName()).log(Level.SEVERE, null, ex);
        } finally {
            try {
                //session结束
                session.shutdown();
            } catch (REConnectionException ex) {
                java.util.logging.Logger.getLogger(SenderDemo.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}
